package mq

import (
	"encoding/json"
	"time"

	"github.com/zeromicro/go-zero/core/threading"

	"gitlab.vspncd.com/mic/common/tools"
)

// 进程内的延迟队列，一般处理1秒以内的延迟
type (
	PlcMqConsume func(topic string, body []byte)

	Worker interface {
		// 启动客户端
		Start()
		// 停止客户端
		Stop()
		// 定时，delayMilli单位毫秒,最低100ms，消费回调有一定的误差，误差不超过100ms
		At(topic string, v interface{}, delayMilli int64)
		// 注册消费函数
		Consume(topic string, consume PlcMqConsume)
	}

	Msg struct {
		Body   []byte
		Delay  int64
		topic  string
		delFlg bool
	}

	defaultLc struct {
		close chan interface{}
		end   chan interface{}
		dChan chan *Msg
		data  []*Msg

		handlers map[string]PlcMqConsume
	}
)

func NewWorker() Worker {
	return &defaultLc{
		close:    make(chan interface{}),
		end:      make(chan interface{}),
		dChan:    make(chan *Msg, 5),
		data:     make([]*Msg, 0, 10),
		handlers: map[string]PlcMqConsume{},
	}
}

func (b *defaultLc) At(topic string, v interface{}, delayMilli int64) {
	body, err := json.Marshal(v)
	if err != nil {
		return
	}

	threading.RunSafe(func() {
		b.dChan <- &Msg{
			Body:  body,
			Delay: time.Now().UnixMilli() + delayMilli,
			topic: topic,
		}
	})
}

func (b *defaultLc) Consume(topic string, consume PlcMqConsume) {
	b.handlers[topic] = consume
}

func (b *defaultLc) Start() {
	go func() {
		interval := 100 * time.Millisecond
		timer := time.NewTimer(interval)

		dealFn := func(isClose bool) {
			timer.Stop()
			cru := time.Now().UnixMilli()
			for _, d := range b.data {
				if !isClose {
					// 关闭的时候，需要立马触发，尽最大可能的让程序不丢消息（可能这会影响精度）
					if d.Delay > cru {
						continue
					}
				}

				if h, ok := b.handlers[d.topic]; ok {
					h(d.topic, d.Body)
				}

				d.delFlg = true
			}

			// 删除处理过的数据
			b.data = tools.DelSliceItem(b.data, func(index int) bool {
				return b.data[index].delFlg
			}).([]*Msg)

			timer.Reset(interval)
		}

		for {
			select {
			case _ = <-b.close:
				dealFn(true)
				b.end <- struct{}{}
				return
			case <-timer.C:
				dealFn(false)
			case d := <-b.dChan:
				if d != nil { // 关闭chan的时候会写入一个nil，这里要过滤
					b.data = append(b.data, d)
				}
			}
		}
	}()
}

func (b *defaultLc) Stop() {
	close(b.dChan)
	b.close <- struct{}{}
	select {
	case _ = <-b.end:
		break
	}
	close(b.close)
	close(b.end)
}
